package com.fanco.learn_datastruct_leetcode.juc;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author qiaowang 2021/8/19
 * 描述： 通过 阻塞队列实现不用手动加锁的 生产者 消费者模式
 */
public class ProductConsumerDemo_2 {
    public static void main(String[] args) {

        MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "\t 生产线程启动");
            System.out.println("");
            System.out.println("");
            try {
                myResource.prod();
                System.out.println("");
                System.out.println("");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "prod").start();


        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "\t 消费线程启动");

            try {
                myResource.consumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "consumer").start();

        // 5秒后，停止生产和消费
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("");
        System.out.println("");
        System.out.println("5秒中后，生产和消费线程停止，线程结束");
        myResource.stop();

    }
}

/**
 * 资源类
 */
class MyResource {
    /**
     * 是否开启生产消费模式
     */
    private volatile boolean flag = true;

    /**
     * 使用原子类,做数据的+-
     */
    private AtomicInteger atomicInteger = new AtomicInteger();

    BlockingQueue<String> blockingQueue = null;

    public MyResource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    /**
     * 生产功能
     */
    public void prod() {
        String data = null;
        boolean returnValue;
        // 只有允许生产时，才能生产
        while (flag) {
            data = atomicInteger.incrementAndGet() + "";
            try {
                // 延迟2s存入一个data
                returnValue = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
                if (returnValue) {
                    System.out.println(Thread.currentThread().getName() + "\t 插入队列成功");
                } else {
                    System.out.println(Thread.currentThread().getName() + "\t 插入队列失败");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        System.out.println(Thread.currentThread().getName() + "\t 生产停止");
    }

    /**
     * 消费功能
     */
    public void consumer() {
        String returnValue;
        while (flag) {
            try {
                returnValue = blockingQueue.poll(2L, TimeUnit.SECONDS);
                if (null == returnValue || returnValue.equalsIgnoreCase("")) {
                    flag = false;
                    System.out.println("超过两秒钟没有消费，停止生产");
                    return;
                }
                System.out.println(Thread.currentThread().getName() + "\t 消费成功");

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void stop() {
        this.flag = false;
    }
}
